﻿#include "tk_net_udp.h"
#include <QDebug>

#include "tk_thread.h"

#include <cstdlib>
#include <functional>
#include <iostream>
#include <boost/asio/buffer.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>

using boost::asio::ip::udp;
using std::placeholders::_1;
using std::placeholders::_2;

namespace tk{
namespace io{

	//----------------------------------------------------------------------

	//
	// This class manages socket timeouts by running the io_context using the timed
	// io_context::run_for() member function. Each asynchronous operation is given
	// a timeout within which it must complete. The socket operations themselves
	// use std::bind to specify the completion handler:
	//
	//   +---------------+
	//   |               |
	//   |    receive    |
	//   |               |
	//   +---------------+
	//           |
	//  async_-  |    +----------------+
	// receive() |    |                |
	//           +--->| handle_receive |
	//                |                |
	//                +----------------+
	//
	// For a given socket operation, the client object runs the io_context to block
	// thread execution until the operation completes or the timeout is reached. If
	// the io_context::run_for() function times out, the socket is closed and the
	// outstanding asynchronous operation is cancelled.
	//
	class client
	{
	public:
		client(boost::asio::io_context * p_io_context, udp::socket *p_socket)
			:p_io_context_(p_io_context),
			p_socket_(p_socket)
		{
		}

		std::size_t receive(const boost::asio::mutable_buffer& buffer,
			std::chrono::steady_clock::duration timeout,
			boost::system::error_code& error)
		{
			// Start the asynchronous operation. The handle_receive function used as a
			// callback will update the error and length variables.
			std::size_t length = 0;
			p_socket_->async_receive(boost::asio::buffer(buffer),
				std::bind(&client::handle_receive, _1, _2, &error, &length));

			// Run the operation until it completes, or until the timeout.
			run(timeout);
			return length;
		}

	private:
		void run(std::chrono::steady_clock::duration timeout)
		{
			// Restart the io_context, as it may have been left in the "stopped" state
			// by a previous operation.
			p_io_context_->restart();

			// Block until the asynchronous operation has completed, or timed out. If
			// the pending asynchronous operation is a composed operation, the deadline
			// applies to the entire operation, rather than individual operations on
			// the socket.
			p_io_context_->run_for(timeout);

			// If the asynchronous operation completed successfully then the io_context
			// would have been stopped due to running out of work. If it was not
			// stopped, then the io_context::run_for call must have timed out.
			if (!p_io_context_->stopped())
			{
				// Cancel the outstanding asynchronous operation.
				p_socket_->cancel();

				// Run the io_context again until the operation completes.
				p_io_context_->run();
			}
		}

		static void handle_receive(
			const boost::system::error_code& error, std::size_t length,
			boost::system::error_code* out_error, std::size_t* out_length)
		{
			*out_error = error;
			*out_length = length;
		}

	private:
		boost::asio::io_context * p_io_context_;
		udp::socket * p_socket_;
	};


TkNetUdp::TkNetUdp()
{
	m_p_io_context = TK_NULL_PTR;
    m_p_socket = TK_NULL_PTR;
    m_p_client = TK_NULL_PTR;
}

TkNetUdp::~TkNetUdp()
{
    CloseAll();
}

tk_bool TkNetUdp::Init(QString remote_ip, tk_uint16 remote_port,tk_uint16 local_port)
{
    TkMutexLocker locker(m_mutex_recv);
	TkMutexLocker locker_(m_mutex_send);
	try
    {
        CloseAll();

        m_local_port = local_port;
        m_remote_ip = remote_ip;
        m_remote_port = remote_port;

        m_listen_endpoint = udp::endpoint(udp::v4(), m_local_port);
        m_remote_endpoint = udp::endpoint(boost::asio::ip::address::from_string(m_remote_ip.toStdString()),
                                          m_remote_port);

		//m_p_io_service = new boost::asio::io_service;
		m_p_io_context = new boost::asio::io_context;
        m_p_socket = new udp::socket(*m_p_io_context, m_listen_endpoint);

        m_p_client = new client(m_p_io_context,m_p_socket);

        udp::socket::receive_buffer_size _rbs(4*1024*1024);
        m_p_socket->set_option(_rbs);

    }
    catch (std::exception& e)
    {
        qDebug()<<"TkNetUdp exception : "<<m_remote_port<<" "<<e.what();
		CloseAll();
        return TK_FALSE;
    }
    return TK_TRUE;
}

void TkNetUdp::Exit()
{
    CloseAll();
}

void TkNetUdp::CloseAll()
{
	TkMutexLocker locker(m_mutex_recv);
	TkMutexLocker locker_(m_mutex_send);

	if (m_p_client)
	{
		delete m_p_client;
		m_p_client = TK_NULL_PTR;
	}
    if(m_p_socket)
    {
        if(m_p_socket->is_open())
        {
            m_p_socket->close();
        }
    }
	if (m_p_socket)
	{
		delete m_p_socket;
		m_p_socket = TK_NULL_PTR;
	}
	if (m_p_io_context)
	{
		delete m_p_io_context;
		m_p_io_context = TK_NULL_PTR;
	}
}

tk_bool TkNetUdp::Send(QByteArray &data)
{
    TkMutexLocker locker(m_mutex_send);

    if(!m_p_socket)
    {
        return TK_FALSE;
    }
    try
    {
        boost::system::error_code ignored_error;
        tk_size_t len = m_p_socket->send_to(
                    boost::asio::buffer(data.data(),static_cast<tk_size_t>(data.size())),
                    m_remote_endpoint, 0, ignored_error);
		
		if (len != data.size())
		{
			return TK_FALSE;
		}
    }
    catch (std::exception& e)
    {
        qDebug()<<"TkNetUdp Send exception : "<<m_remote_port<<" "<<e.what();
        return TK_FALSE;
    }
    return TK_TRUE;
}

tk_bool TkNetUdp::Recv(QByteArray &data,tk_int32 timeout_ms)
{
	TkMutexLocker locker(m_mutex_recv);
	if (!m_p_socket)
    {
        return TK_FALSE;
    }
    try
    {
        boost::system::error_code ec;

        if (m_p_socket/* && m_p_socket->available()>0*/)
        {
             tk_size_t len = m_p_client->receive(boost::asio::buffer(m_recv_buf),
												std::chrono::milliseconds(timeout_ms),
                                                 ec);
             if(!ec)
             {
				 if (len)
				 {
					 data.resize(static_cast<tk_int32>(len));
					 memcpy(data.data(), m_recv_buf.data(), sizeof(tk_uint8)*len);
					 return TK_TRUE;
				 }
				 else
				 {
					 return TK_FALSE;
				 }
             }
             else
             {
                 //qDebug()<<"TkNetUdp Receive error: "<<m_remote_port <<" "<<ec.message().c_str();
                 return TK_FALSE;
             }
        }
        else
        {
            return TK_FALSE;
        }
    }
    catch (std::exception& e)
    {
        qDebug()<<"TkNetUdp Send exception : "<<m_remote_port<<" "<<e.what();
        return TK_FALSE;
    }
}

}//namespace io
}//namespace tk
